#pragma once
#include <muduo/net/TcpServer.h>     //调用muduo库中的TcpServer类
#include <muduo/net/EventLoop.h>     //调用muduo库中的EventLoop类
#include <muduo/net/TcpConnection.h> //调用muduo库中的TcpConnection类
#include <muduo/net/Buffer.h>        //调用muduo库中的Buffer类
#include <muduo/net/TcpClient.h>
#include <muduo/net/EventLoop.h>
#include <muduo/net/EventLoopThread.h>
#include <muduo/base/CountDownLatch.h>

#include <mutex>
#include <unordered_map>

#include "Detail.hpp"
#include "Fileds.hpp"
#include "Message.hpp"
#include "Abstract.hpp"

namespace zgwrpc
{
    class MuduoBuffer : public BaseBuffer
    {
    public:
        using ptr = std::shared_ptr<MuduoBuffer>;
        MuduoBuffer(muduo::net::Buffer *buffer) : _buffer(buffer)
        {
        }
        virtual ~MuduoBuffer() {}
        virtual size_t readableSize() override // 返回可读字节数
        {
            return _buffer->readableBytes();
        }
        virtual int32_t peekInt32() override //  跳过字节并返回int32_t类型数据
        {
            return _buffer->peekInt32();
        }
        virtual void retrieveInt32() override // 跳过32字节
        {
            _buffer->retrieveInt32();
        }
        virtual int32_t readInt32() override // 读取int32_t类型数据
        {
            return _buffer->readInt32();
        }
        virtual std::string retrievAsString(size_t len) override // 返回指定长度的字符串，并跳过相应字节
        {
            return _buffer->retrieveAsString(len);
        }

    private:
        muduo::net::Buffer *_buffer;
    };

    class BufferFactory
    {
    public:
        static BaseBuffer::ptr create(muduo::net::Buffer *buffer)
        {
            return std::make_shared<MuduoBuffer>(buffer);
        }
    };

    class LVProtocol : public BaseProtocol
    {
    public:
        // |--len--|--value--|           第一个字段是数据长度
        // |--len--|--mtype--|--idlen--|--id--|--body--|            第二个字段是数据值
        using ptr = std::shared_ptr<LVProtocol>;
        LVProtocol() {}
        virtual ~LVProtocol() override {}
        virtual bool canProcessed(const BaseBuffer::ptr &buf) override // 检查是否可以处理数据
        {
            if (buf->readableSize() < _Messagelen)
            {
                return false;
            }
            int32_t total_len = buf->peekInt32();
            LOG(LogLevel::INFO) << "total len" << total_len;
            if (buf->readableSize() < (total_len + _Messagelen)) // 如果数据长度+4(len的大小)大于可读字节数，则不能处理
                return false;
            return true;
        }
        virtual bool onMessage(BaseMessage::ptr &msg, const BaseBuffer::ptr &buf) override // 处理消息
        {
            // 调用onMessageImpl处理消息时默认认为缓冲区长度一定是能够组成一条消息的
            int32_t total_len = buf->readInt32();  // 获取数据长度
            MType mtype = (MType)buf->readInt32(); // 获取消息类型
            int32_t idlen = (buf->readInt32());    // 获取id长度

            string id = buf->retrievAsString(idlen); // 获取id

            int32_t body_len = total_len - _Mtypelen - _Idlenlen - idlen;
            LOG(LogLevel::INFO) << "total len" << total_len;
            LOG(LogLevel::INFO) << "mtype len" << _Mtypelen;
            LOG(LogLevel::INFO) << "idlen" << idlen;
            LOG(LogLevel::INFO) << "body len" << body_len;
            string body = buf->retrievAsString(body_len); // 获取消息体

            msg = MessageFactory::createMessage(mtype);
            if (msg == nullptr)
            {
                LOG(LogLevel::ERROR) << "消息类型错误，构造消息失败";
                return false;
            }

            bool ret = msg->unSerialize(body);
            if (ret == false)
            {
                LOG(LogLevel::ERROR) << "消息体反序列化失败";
                return false;
            }
            msg->setId(id);
            msg->setMType(mtype);
            return true;
        }
        virtual std::string serialize(const BaseMessage::ptr &msg) override // 序列化消息
        {
            // |--len--|--mtype--|--idlen--|--id--|--body--|           序列化格式
            std::string body = msg->serialize();          // 获取消息体
            std::string id = msg->rid();                  // 获取id
            int32_t mtype = htonl((int32_t)msg->mtype()); // 转换消息类型为网络字节序
            int32_t idlen = htonl(id.size());             // 转换id长度为网络字节序,以便数据传输的跨平台性

            int32_t total_len = htonl(_Mtypelen + _Idlenlen + id.size() + body.size()); // 计算消息总长度并转换为网络字节序
            string result;
            result.append((char *)&total_len, sizeof(total_len)); // 写入消息长度
            result.append((char *)&mtype, sizeof(mtype));         // 写入消息类型
            result.append((char *)&idlen, sizeof(idlen));         // 写入id长度
            result.append(id);                                    // 写入id
            result.append(body);                                  // 写入消息体
            return result;
        }

    private:
        const size_t _Messagelen = 4; // 4字节的消息长度
        const size_t _Mtypelen = 4;   // 4字节的消息类型
        const size_t _Idlenlen = 4;   // 4字节的id长度
    };

    class ProtocolFactory
    {
    public:
        template <typename... Args>
        static BaseProtocol::ptr create(Args &&...args)
        {
            return std::make_shared<LVProtocol>(std::forward<Args>(args)...);
        }
    };

    class MuduoConnection : public BaseConnection
    {
    public:
        using ptr = std::unique_ptr<MuduoConnection>;
        MuduoConnection(muduo::net::TcpConnectionPtr con, const BaseProtocol::ptr &protocol)
            : _protocol(protocol), _con(con)
        {
        }
        virtual ~MuduoConnection() {};

        virtual void send(const BaseMessage::ptr &msg) override
        {
            std::string body = _protocol->serialize(msg);
            _con->send(body);
        }
        virtual void shutdown() override
        {
            _con->shutdown();
        }
        virtual bool connected() override
        {
            return _con->connected();
        }
        virtual void reset() override
        {
            _con.reset();
        }

    private:
        BaseProtocol::ptr _protocol;
        muduo::net::TcpConnectionPtr _con;
    };

    class ConnectionFactory
    {
    public:
        template <typename... Args>
        static BaseConnection::ptr create(Args &&...args)
        {
            return std::make_shared<MuduoConnection>(std::forward<Args>(args)...);
        }
    };

    class MuduoServer : public BaseServer
    {
    public:
        // using ConnectionCallback = std::function<void(const BaseConnection::ptr &conn)>;
        // using MessageCallback = std::function<void(const BaseConnection::ptr &conn, const BaseMessage::ptr &buf)>;
        // using CloseCallback = std::function<void(const BaseConnection::ptr &conn)>;
        using ptr = std::shared_ptr<BaseServer>;
        MuduoServer(int port) : _server(&_loop, muduo::net::InetAddress("0.0.0.0", port), "MuduoServer")
        {
            _protocol = ProtocolFactory::create();
        }

        virtual ~MuduoServer() {}
        virtual void start() override // 启动服务
        {
            _server.setConnectionCallback(std::bind(&MuduoServer::OnConnection, this, std::placeholders::_1));
            _server.setMessageCallback(std::bind(&MuduoServer::OnMessage, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _server.start();
            _loop.loop();
        }

        void OnConnection(const muduo::net::TcpConnectionPtr &conn) // 连接时的回调函数
        {
            if (conn->connected())
            {
                LOG(LogLevel::INFO) << "连接成功";
                auto muduo_conne = ConnectionFactory::create(conn, _protocol); // 创建muduo连接
                {
                    unique_lock<mutex> lock(_mutex);
                    _conns.insert(std::make_pair(conn, muduo_conne));
                }
                if (_connectCallback)
                    _connectCallback(muduo_conne); // 调用连接回调函数
            }
            else
            {
                LOG(LogLevel::INFO) << "连接断开";
                BaseConnection::ptr muduo_conne;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto it = _conns.find(conn);
                    if (it != _conns.end())
                    {
                        return;
                    }
                    muduo_conne = it->second;
                    _conns.erase(conn);
                }
                if (_closeCallback)
                    _closeCallback(muduo_conne); // 调用关闭回调函数
            }
        }

        void OnMessage(const muduo::net::TcpConnectionPtr &conn, muduo::net::Buffer *buf, muduo::Timestamp time) // 接收到消息时的回调函数
        {
            LOG(LogLevel::INFO) << "接收到新消息";
            auto base_buf = BufferFactory::create(buf); // 创建muduo缓冲区

            while (1)
            {
                if (_protocol->canProcessed(base_buf) == false)
                {
                    if (base_buf->readableSize() > maxDataSize)
                    {
                        conn->shutdown();
                        LOG(LogLevel::ERROR) << "数据包过大";
                        return;
                    }
                    LOG(LogLevel::INFO) << "数据包不能处理";
                    break;
                }
                BaseConnection::ptr base_conne;
                {
                    unique_lock<mutex> lock(_mutex);
                    auto it = _conns.find(conn);
                    if (it == _conns.end())
                    {
                        conn->shutdown();
                        LOG(LogLevel::ERROR) << "连接不存在";
                        return;
                    }
                    base_conne = it->second;
                }
                BaseMessage::ptr msg;
                bool ret = _protocol->onMessage(msg, base_buf);
                if (ret == false)
                {
                    conn->shutdown();
                    LOG(LogLevel::ERROR) << "消息解析失败";
                    return;
                }
                if (_messageCallback)
                    _messageCallback(base_conne, msg); // 调用消息回调函数
            }
        }

    private:
        const size_t maxDataSize = (1 << 16);
        BaseProtocol::ptr _protocol;
        muduo::net::EventLoop _loop;                                                  // muduo库中的EventLoop类，出发IO事件,用于事件循环
        muduo::net::TcpServer _server;                                                // muduo库中的TcpServer类，接收连接请求
        std::mutex _mutex;                                                            // 互斥锁
        std::unordered_map<muduo::net::TcpConnectionPtr, BaseConnection::ptr> _conns; // 存放连接信息
    };

    class ServerFactory
    {
    public:
        template <typename... Args>
        static BaseServer::ptr create(Args &&...args)
        {
            return std::make_shared<MuduoServer>(std::forward<Args>(args)...);
        }
    };

    class MuduoClient : public BaseClient
    {
    public:
        using ptr = std::shared_ptr<MuduoClient>;

        MuduoClient(int port, string sip) : _loop(_loopThread.startLoop()),
                                            _client(_loop, muduo::net::InetAddress(sip, port), "MuduoClient"),
                                            _latch(1),
                                            _protocol(ProtocolFactory::create())
        {
            connect();
            _client.connect();
            _latch.wait();
        }

        void OnConnection(const muduo::net::TcpConnectionPtr &conn)
        {
            if (conn->connected() == true)
            {
                _latch.countDown(); // 计数--，为
                _con = ConnectionFactory::create(conn, _protocol);
            }
            else
            {
                _con->reset(); // 重置连接
                std::cout << "Disconnected" << std::endl;
            }
        }
        void OnMessage(const muduo::net::TcpConnectionPtr &conn, muduo::net::Buffer *buf, muduo::Timestamp time)
        {
            LOG(LogLevel::INFO) << "接收到新消息";
            auto base_buf = BufferFactory::create(buf); // 创建muduo缓冲区

            while (1)
            {
                if (_protocol->canProcessed(base_buf) == false)
                {
                    if (base_buf->readableSize() > maxDataSize)
                    {
                        conn->shutdown();
                        LOG(LogLevel::ERROR) << "数据包过大";
                        return;
                    }
                    LOG(LogLevel::INFO) << "数据包不能处理";
                    break;
                }
                BaseMessage::ptr msg;
                bool ret = _protocol->onMessage(msg, base_buf);
                if (ret == false)
                {
                    conn->shutdown();
                    LOG(LogLevel::ERROR) << "消息解析失败";
                    return;
                }
                LOG(LogLevel::INFO) << "进行回调";
                if (_messageCallback)
                    _messageCallback(_con, msg); // 调用消息回调函数
            }
        }

        virtual void connect() override // 连接服务
        {
            _client.setConnectionCallback(std::bind(&MuduoClient::OnConnection, this, std::placeholders::_1));
            _client.setMessageCallback(std::bind(&MuduoClient::OnMessage, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
        }
        virtual bool isConnected() override // 是否连接成功
        {
            return (_con && _con->connected());
        }

        virtual void send(const BaseMessage::ptr &msg) override // 发送消息
        {
            LOG(LogLevel::INFO) << "发送消息";
            _con->send(msg);
        }
        virtual void shutdown() override // 关闭连接
        {
            _client.disconnect();
        }
        virtual BaseConnection::ptr connection() override
        {
            return _con;
        }

    private:
        const size_t maxDataSize = (1 << 16);
        BaseProtocol::ptr _protocol;
        BaseConnection::ptr _con;
        muduo::net::EventLoopThread _loopThread;
        muduo::net::EventLoop *_loop;
        muduo::CountDownLatch _latch;
        muduo::net::TcpClient _client;
    };

    class ClientFactory
    {
    public:
        template <typename... Args>
        static BaseClient::ptr create(Args &&...args)
        {
            return make_shared<MuduoClient>(std::forward<Args>(args)...);
        }

    private:
    };
}